/*
 * The MIT License
 *
 * Copyright 2014 David Rodenas Pico.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 */

package edu.uoc.sd.fr14.comp;

import edu.uoc.sd.fr14.mesg.Message;
import edu.uoc.sd.fr14.mesg.TargetId;
import java.util.Timer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @author drodenas
 */
public abstract class Component implements Bus.Listener {
    private final static Message EOF = new Message() { @Override public TargetId getTarget() {return null;} };
    
    
    private static final ExecutorService executors = Executors.newCachedThreadPool();

    /**
     * Stops all active threads in the executor thread pool
     * instead of waiting 60 seconds to finish them naturally.
     */
    public static void shutdown() {
        executors.shutdown();
    }

    private Bus bus;
    private final ConcurrentMap<String,Processor> processors;
    private final BlockingQueue<Message> queue;
    private final AtomicReference<Timer> timer;

    public Component(String name) {
        this.processors = new ConcurrentHashMap<>();
        this.queue =      new LinkedBlockingQueue<>();
        this.timer =      new AtomicReference<>();
        
        createProcessor(name, new Runnable() {
            @Override
            public void run() {
                Message message = takeMessage();
                if (message == null) return;
                process(message);
            }
        });
    }
    
    protected abstract void process(Message message);

    @Override
    public final void execute(Message message) {
        try { queue.put(message); } 
        catch (InterruptedException ex) { throw new RuntimeException("Unexpected exception. May be stopping?", ex); }
    }

    @Override
    public final void setBus(Bus bus) {
        this.bus = bus;
        bus.addListener(this);
    }
    
    public void stop() {
        for (Processor processor : processors.values()) {
            processor.stop();
        }
        queue.clear();
        queue.add(EOF);
        getTimer().cancel();
    }
    
    protected final void createProcessor(String processorName, Runnable task) {
        if (processorName == null) throw new NullPointerException("Illegal argument: processorName cannot be null");
        if (task == null) throw new NullPointerException("Illegal argument: task cannot be null");
        
        if (processors.putIfAbsent(processorName, new Processor(task)) != null) {
            throw new IllegalStateException("Illegal argument: processorName already created");
        }
    }
    
    protected void deliver(Message message) {
        if (message == null) throw new NullPointerException("Illegal argument: message cannot be null");
        
        Bus current = this.bus;
        if (current != null) {
            current.deliver(message);
        }
    }
    
    protected Timer getTimer() {
        Timer current = timer.get();
        if (current == null) {
            current = new Timer("component-timer", true);
            boolean changed = timer.compareAndSet(null, current);
            if (!changed) {
                current.cancel();
                current = timer.get();
            }
        }
        
        return current;
    }

    protected void stopProcessor(String processorName) {
        if (processorName == null) throw new NullPointerException("Illegal argument: inputName cannot be null");
        
        Processor processor = processors.remove(processorName);
        if (processor != null) {
            processor.stop();
        }
    }
    
    /**
     *   @return the next message (waiting as needed) or null if stop was invoked
     */
    protected final Message takeMessage() {
        try {
            Message message = queue.take();
            if (message == EOF) return null;
            else return message;
        }
        catch (InterruptedException ex) { throw new RuntimeException("Unexpected exception. May be stopping?", ex); }
    }
    
    private static class Processor {
        
        private volatile boolean stopped;

        private Processor(final Runnable task) {
            executors.execute(new Runnable() {
                @Override
                public void run() {
                    while (!stopped) {
                        task.run();
                    }
                }
            });
        }
        
        private void stop() {
            stopped = true;
        }
        
    }
    
    
}
